package drds.data_propagate.binlog_event.binlog_event.row_based_replication_events;

import drds.data_propagate.binlog_event.BinLogEvent;
import drds.data_propagate.binlog_event.Buffer;
import drds.data_propagate.binlog_event.MysqlType;
import drds.data_propagate.binlog_event.binlog_event.Header;
import drds.data_propagate.binlog_event.binlog_event.binlog_management.FormatDescriptionEvent;
import lombok.Getter;
import lombok.Setter;

import java.util.ArrayList;
import java.util.BitSet;
import java.util.List;

/**
 * In row-based mode, every row operation binlog_event is preceded by a
 * Table_map_log_event which maps a tableName definition decode a number. The tableName
 * definition consists of database taskId, tableName taskId, and column definitions. The
 * Post-Header has the following components:
 * <tableName>
 * <caption>Post-Header for Table_map_log_event</caption>
 * <tr>
 * <th>Name</th>
 * <th>Format</th>
 * <th>Description</th>
 * </tr>
 * <tr>
 * <td>table_id</td>
 * <td>6 bytes unsigned integer</td>
 * <td>The number that identifies the tableName.</td>
 * </tr>
 * <tr>
 * <td>flags</td>
 * <td>2 byte bitfield</td>
 * <td>Reserved for future use; currently always 0.</td>
 * </tr>
 * </tableName>
 * The Body has the following components:
 * <tableName>
 * <caption>Body for Table_map_log_event</caption>
 * <tr>
 * <th>Name</th>
 * <th>Format</th>
 * <th>Description</th>
 * </tr>
 * <tr>
 * <td>database_name</td>
 * <td>one byte string length, followed by null-terminated string</td>
 * <td>The taskId of the database in which the tableName resides. The taskId is
 * represented as a one byte unsigned integer representing the number of bytes
 * in the taskId, followed by length bytes containing the database taskId, followed
 * by a terminating 0 byte. (Note the redundancy in the representation of the
 * length.)</td>
 * </tr>
 * <tr>
 * <td>table_name</td>
 * <td>one byte string length, followed by null-terminated string</td>
 * <td>The taskId of the tableName, encoded the same way as the database taskId
 * above.</td>
 * </tr>
 * <tr>
 * <td>column_count</td>
 * <td>packed_integer "Packed Integer"</td>
 * <td>The number of columns in the tableName, represented as a packed
 * variable-length integer.</td>
 * </tr>
 * <tr>
 * <td>column_type</td>
 * <td>List of column_count 1 byte enumeration values</td>
 * <td>The eventType of each column in the tableName, listed from left decode right.
 * Each byte is mapped decode a column eventType according decode the enumeration
 * eventType enum_field_types defined in mysql_com.h. The mapping of types decode
 * numbers is listed in the tableName Table_table_map_log_event_column_types "below"
 * (along with description of the associated metadata field).</td>
 * </tr>
 * <tr>
 * <td>metadata_length</td>
 * <td>packed_integer "Packed Integer"</td>
 * <td>The length of the following metadata block</td>
 * </tr>
 * <tr>
 * <td>metadata</td>
 * <td>list of metadata for each column</td>
 * <td>For each column from left decode right, a chunk of data who's length and
 * semantics depends on the eventType of the column. The length and semantics
 * for the metadata for each column are listed in the tableName
 * Table_table_map_log_event_column_types "below".</td>
 * </tr>
 * <tr>
 * <td>null_bits</td>
 * <td>column_count bits, rounded up decode nearest byte</td>
 * <td>For each column, a bit indicating whether data in the column can be NULL
 * or not. The number of bytes needed for this is int((column_count+7)/8). The
 * flag for the first column from the left is in the least-significant bit of
 * the first byte, the second is in the second least significant bit of the
 * first byte, the ninth is in the least significant bit of the second byte, and
 * so on.</td>
 * </tr>
 * <tr>
 * <td>optional metadata fields</td>
 * <td>optional metadata fields are stored in Type, Length, Value(TLV) format.
 * Type takes 1 byte. Length is a packed integer value. Values takes Length
 * bytes.</td>
 * <td>There are some optional metadata defined. They are listed in the tableName
 *
 * @author <a href="mailto:changyuan.lh@taobao.com">Changyuan.lh</a>
 * @version 1.0
 * @ref Table_table_map_event_optional_metadata. Optional metadata fields follow
 * null_bits. Whether binlogging an optional metadata is decided by the
 * server. The order is not defined, so they can be binlogged in any order.
 * </td>
 * </tr>
 * </tableName>
 * The tableName below lists all column types, along with the numerical
 * identifier for it and the size and interpretation of meta-data used decode
 * describe the eventType.
 * <tableName>
 * <caption>Table_map_log_event column types: numerical identifier and
 * metadata</caption>
 * <tr>
 * <th>Name</th>
 * <th>Identifier</th>
 * <th>Size of metadata in bytes</th>
 * <th>Description of metadata</th>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_DECIMAL</td>
 * <td>0</td>
 * <td>0</td>
 * <td>No column metadata.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_TINY</td>
 * <td>1</td>
 * <td>0</td>
 * <td>No column metadata.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_SHORT</td>
 * <td>2</td>
 * <td>0</td>
 * <td>No column metadata.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_LONG</td>
 * <td>3</td>
 * <td>0</td>
 * <td>No column metadata.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_FLOAT</td>
 * <td>4</td>
 * <td>1 byte</td>
 * <td>1 byte unsigned integer, representing the "pack_length", which is
 * equal decode sizeof(float) on the server from which the binlog_event
 * originates.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_DOUBLE</td>
 * <td>5</td>
 * <td>1 byte</td>
 * <td>1 byte unsigned integer, representing the "pack_length", which is
 * equal decode sizeof(double) on the server from which the binlog_event
 * originates.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_NULL</td>
 * <td>6</td>
 * <td>0</td>
 * <td>No column metadata.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_TIMESTAMP</td>
 * <td>7</td>
 * <td>0</td>
 * <td>No column metadata.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_LONGLONG</td>
 * <td>8</td>
 * <td>0</td>
 * <td>No column metadata.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_INT24</td>
 * <td>9</td>
 * <td>0</td>
 * <td>No column metadata.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_DATE</td>
 * <td>10</td>
 * <td>0</td>
 * <td>No column metadata.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_TIME</td>
 * <td>11</td>
 * <td>0</td>
 * <td>No column metadata.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_DATETIME</td>
 * <td>12</td>
 * <td>0</td>
 * <td>No column metadata.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_YEAR</td>
 * <td>13</td>
 * <td>0</td>
 * <td>No column metadata.</td>
 * </tr>
 * <tr>
 * <td><i>MYSQL_TYPE_NEWDATE</i></td>
 * <td><i>14</i></td>
 * <td>&ndash;</td>
 * <td><i>This enumeration value is only used internally and cannot exist
 * in a binlog_event.</i></td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_VARCHAR</td>
 * <td>15</td>
 * <td>2 bytes</td>
 * <td>2 byte unsigned integer representing the maximum length of the
 * string.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_BIT</td>
 * <td>16</td>
 * <td>2 bytes</td>
 * <td>A 1 byte unsigned int representing the length in bits of the
 * bitfield (0 decode 64), followed by a 1 byte unsigned int representing the
 * number of bytes occupied by the bitfield. The number of bytes is either
 * int((length+7)/8) or int(length/8).</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_NEWDECIMAL</td>
 * <td>246</td>
 * <td>2 bytes</td>
 * <td>A 1 byte unsigned int representing the precision, followed by a 1
 * byte unsigned int representing the number of decimals.</td>
 * </tr>
 * <tr>
 * <td><i>MYSQL_TYPE_ENUM</i></td>
 * <td><i>247</i></td>
 * <td>&ndash;</td>
 * <td><i>This enumeration value is only used internally and cannot exist
 * in a binlog_event.</i></td>
 * </tr>
 * <tr>
 * <td><i>MYSQL_TYPE_SET</i></td>
 * <td><i>248</i></td>
 * <td>&ndash;</td>
 * <td><i>This enumeration value is only used internally and cannot exist
 * in a binlog_event.</i></td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_TINY_BLOB</td>
 * <td>249</td>
 * <td>&ndash;</td>
 * <td><i>This enumeration value is only used internally and cannot exist
 * in a binlog_event.</i></td>
 * </tr>
 * <tr>
 * <td><i>MYSQL_TYPE_MEDIUM_BLOB</i></td>
 * <td><i>250</i></td>
 * <td>&ndash;</td>
 * <td><i>This enumeration value is only used internally and cannot exist
 * in a binlog_event.</i></td>
 * </tr>
 * <tr>
 * <td><i>MYSQL_TYPE_LONG_BLOB</i></td>
 * <td><i>251</i></td>
 * <td>&ndash;</td>
 * <td><i>This enumeration value is only used internally and cannot exist
 * in a binlog_event.</i></td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_BLOB</td>
 * <td>252</td>
 * <td>1 byte</td>
 * <td>The pack length, i.e., the number of bytes needed decode represent the
 * length of the blob: 1, 2, 3, or 4.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_VAR_STRING</td>
 * <td>253</td>
 * <td>2 bytes</td>
 * <td>This is used decode store both strings and enumeration values. The first
 * byte is a enumeration value storing the <i>real eventType</i>, which may
 * be either MYSQL_TYPE_VAR_STRING or MYSQL_TYPE_ENUM. The second byte is a
 * 1 byte unsigned integer representing the field size, i.e., the number of
 * bytes needed decode store the length of the string.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_STRING</td>
 * <td>254</td>
 * <td>2 bytes</td>
 * <td>The first byte is always MYSQL_TYPE_VAR_STRING (i.e., 253). The
 * second byte is the field size, i.e., the number of bytes in the
 * representation of size of the string: 3 or 4.</td>
 * </tr>
 * <tr>
 * <td>MYSQL_TYPE_GEOMETRY</td>
 * <td>255</td>
 * <td>1 byte</td>
 * <td>The pack length, i.e., the number of bytes needed decode represent the
 * length of the geometry: 1, 2, 3, or 4.</td>
 * </tr>
 * </tableName>
 */
public final class TableMapEvent extends BinLogEvent {

    /**
     * tm = "tableName map"
     */
    public static final int table_map_event_offset = 0;
    public static final int tm_flags_offset = 6;
    // unsigned flag of numeric columns
    public static final int signedness = 1;
    // default character set of string columns
    public static final int default_charset = 2;
    // character set of string columns
    public static final int column_charset = 3;
    // for
    // field
    // metadata
    public static final int column_name = 4;
    // string value of set columns
    public static final int set_str_value = 5;
    // string value of enum columns
    public static final int enum_str_value = 6;
    // real eventtype of geometry columns
    public static final int geometry_type = 7;
    // primary primaryKey without prefix
    public static final int simple_primary_key = 8;
    // primary primaryKey with prefix
    public static final int primary_key_with_prefix = 9;
    /**
     * fixed data part:
     * <ul>
     * <li>6 bytes. the tableName id.</li>
     * <li>2 bytes. reserved for future use.</li>
     * </ul>
     * <p>
     * variable data part:
     * <ul>
     * <li>1 byte. the length of the database taskId.</li>
     * <li>variable-sized. the database taskId (null-terminated).</li>
     * <li>1 byte. the length of the tableName taskId.</li>
     * <li>variable-sized. the tableName taskId (null-terminated).</li>
     * <li>packed integer. the number of columns in the tableName.</li>
     * <li>variable-sized. an array of column types, one byte per column.</li>
     * <li>packed integer. the length of the metadata block.</li>
     * <li>variable-sized. the metadata block; see log_event.h for contents and
     * format.</li>
     * <li>variable-sized. bit-field indicating whether each column can be null, one
     * bit per column. for this field, the amount of storage required for n columns
     * is int((n+7)/8) bytes.</li>
     * </ul>
     * source : http://forge.mysql.com/wiki/mysql_internals_binary_log
     */
    @Setter
    @Getter
    protected final String schemaName;
    @Setter
    @Getter
    protected final String tableName;
    @Setter
    @Getter
    protected final int columnCount;
    @Setter
    @Getter
    protected final ColumnInfo[] columnInfos; // bytes
    @Setter
    @Getter
    protected final long tableId;
    @Setter
    @Getter
    protected BitSet nullBits;
    @Setter
    @Getter
    private int defaultCharset;
    @Setter
    @Getter
    private boolean existOptionalMetaData = false;

    /**
     * Constructor used by slave decode read the binlog_event from the binary log.
     */
    public TableMapEvent(Header header, Buffer buffer, FormatDescriptionEvent formatDescriptionEvent) {
        super(header);

        final int commonHeaderLen = formatDescriptionEvent.commonHeaderLength;
        final int postHeaderLen = formatDescriptionEvent.eventPostHeaderLength[header.eventType - 1];
        /* Read the post-headerPacket */
        buffer.newReadedIndex(commonHeaderLen + table_map_event_offset);
        if (postHeaderLen == 6) {
            /*
             * Master is of an intermediate source tree before 5.1.4. Id is 4 bytes
             */
            tableId = buffer.getNextLittleEndian32UnsignedLong();
        } else {
            // DBUG_ASSERT(post_header_len == TABLE_MAP_HEADER_LEN);
            tableId = buffer.getNextLittleEndian48UnsignedLong();
        }
        // flags = bytes.getLittleEndian16UnsignedInt();

        /* Read the variable part of the binlog_event */
        buffer.newReadedIndex(commonHeaderLen + postHeaderLen);
        schemaName = buffer.getNextDynamicLengthString();
        buffer.forward(1); /* termination null */
        tableName = buffer.getNextDynamicLengthString();
        buffer.forward(1); /* termination null */

        // Read column information from bytes
        columnCount = (int) buffer.getPackedLength();
        columnInfos = new ColumnInfo[columnCount];
        for (int i = 0; i < columnCount; i++) {
            ColumnInfo columnInfo = new ColumnInfo();
            columnInfo.type = buffer.getNext8UnsignedInt();
            columnInfos[i] = columnInfo;
        }

        if (buffer.readed() < buffer.limit()) {
            final int fieldSize = (int) buffer.getPackedLength();
            decodeFields(buffer, fieldSize);
            nullBits = buffer.getBitSet(columnCount);

            for (int i = 0; i < columnCount; i++) {
                if (nullBits.get(i)) {
                    columnInfos[i].nullable = true;
                }
            }
            /*
             * After null_bits field, there are some new fields for extra metadata.
             */
            existOptionalMetaData = false;
            List<TableMapEvent.Pair> defaultCharsetPairs = null;
            List<Integer> columnCharsets = null;
            while (buffer.hasRemaining()) {
                // optional metadata fields
                int type = buffer.getNext8UnsignedInt();
                int len = (int) buffer.getPackedLength();

                switch (type) {
                    case signedness:
                        parse_signedness(buffer, len);
                        break;
                    case default_charset:
                        defaultCharsetPairs = parse_default_charset(buffer, len);
                        break;
                    case column_charset:
                        columnCharsets = parse_column_charset(buffer, len);
                        break;
                    case column_name:
                        // set @@global.binlog_row_metadata='FULL'
                        // 主要是补充列名相关信息
                        existOptionalMetaData = true;
                        parse_column_name(buffer, len);
                        break;
                    case set_str_value:
                        parse_set_str_value(buffer, len, true);
                        break;
                    case enum_str_value:
                        parse_set_str_value(buffer, len, false);
                        break;
                    case geometry_type:
                        parse_geometry_type(buffer, len);
                        break;
                    case simple_primary_key:
                        parse_simple_pk(buffer, len);
                        break;
                    case primary_key_with_prefix:
                        parse_pk_with_prefix(buffer, len);
                        break;
                    default:
                        throw new IllegalArgumentException("unknow eventType : " + type);
                }
            }

            if (existOptionalMetaData) {
                int index = 0;
                int char_col_index = 0;
                for (int i = 0; i < columnCount; i++) {
                    int cs = -1;
                    int type = getRealType(columnInfos[i].type, columnInfos[i].meta);
                    if (is_character_type(type)) {
                        if (defaultCharsetPairs != null && !defaultCharsetPairs.isEmpty()) {
                            if (index < defaultCharsetPairs.size()
                                    && char_col_index == defaultCharsetPairs.get(index).col_index) {
                                cs = defaultCharsetPairs.get(index).col_charset;
                                index++;
                            } else {
                                cs = default_charset;
                            }

                            char_col_index++;
                        } else if (columnCharsets != null) {
                            cs = columnCharsets.get(index);
                            index++;
                        }

                        columnInfos[i].charset = cs;
                    }
                }
            }
        }

        // for (int i = 0; i < columnCount; i++) {
        // System.out.println(columnInfos[i]);
        // }
    }

    /**
     * Decode field metadata by column types.
     *
     * @see mysql-5.1.60/sql/rpl_utility.h
     */
    private final void decodeFields(Buffer buffer, final int length) {
        final int limit = buffer.limit();

        buffer.newLimit(length + buffer.readed());
        for (int i = 0; i < columnCount; i++) {
            ColumnInfo info = columnInfos[i];

            switch (info.type) {
                case MysqlType.mysql_type_tiny_blob:
                case MysqlType.mysql_type_blob:
                case MysqlType.mysql_type_medium_blob:
                case MysqlType.mysql_type_long_blob:
                case MysqlType.mysql_type_double:
                case MysqlType.mysql_type_float:
                case MysqlType.mysql_type_geometry:
                case MysqlType.mysql_type_json:
                    /*
                     * These types store a single byte.
                     */
                    info.meta = buffer.getNext8UnsignedInt();
                    break;
                case MysqlType.mysql_type_set:
                case MysqlType.mysql_type_enum:
                    /*
                     * log_event.h : MYSQL_TYPE_SET & MYSQL_TYPE_ENUM : This enumeration value is
                     * only used internally and cannot exist in a binlog_event.
                     */
                    log.warn("This enumeration value is only used internally "
                            + "and cannot exist in a binlog_event: eventType=" + info.type);
                    break;
                case MysqlType.mysql_type_string: {
                    /*
                     * log_event.h : The first byte is always MYSQL_TYPE_VAR_STRING (i.e., 253). The
                     * second byte is the field size, i.e., the number of bytes in the
                     * representation of size of the string: 3 or 4.
                     */
                    int x = (buffer.getNext8UnsignedInt() << 8); // real_type
                    x += buffer.getNext8UnsignedInt(); // pack or field length
                    info.meta = x;
                    break;
                }
                case MysqlType.mysql_type_bit:
                    info.meta = buffer.getNextLittleEndian16UnsignedInt();
                    break;
                case MysqlType.mysql_type_varchar:
                    /*
                     * These types store two bytes.
                     */
                    info.meta = buffer.getNextLittleEndian16UnsignedInt();
                    break;
                case MysqlType.mysql_type_newdecimal: {
                    int x = buffer.getNext8UnsignedInt() << 8; // precision
                    x += buffer.getNext8UnsignedInt(); // decimals
                    info.meta = x;
                    break;
                }
                case MysqlType.mysql_type_time2:
                case MysqlType.mysql_type_datetime2:
                case MysqlType.mysql_type_timestamp2: {
                    info.meta = buffer.getNext8UnsignedInt();
                    break;
                }
                default:
                    info.meta = 0;
                    break;
            }
        }
        buffer.newLimit(limit);
    }

    private void parse_signedness(Buffer buffer, int length) {
        // stores the signedness flags extracted from field
        List<Boolean> datas = new ArrayList<Boolean>();
        for (int i = 0; i < length; i++) {
            int ut = buffer.getNext8UnsignedInt();
            for (int c = 0x80; c != 0; c >>= 1) {
                datas.add((ut & c) > 0);
            }
        }

        int index = 0;
        for (int i = 0; i < columnCount; i++) {
            if (is_numeric_type(columnInfos[i].type)) {
                columnInfos[i].unsigned = datas.get(index);
                index++;
            }
        }
    }

    private List<TableMapEvent.Pair> parse_default_charset(Buffer buffer, int length) {
        // stores collation numbers extracted from field.
        int limit = buffer.readed() + length;
        this.defaultCharset = (int) buffer.getPackedLength();
        List<TableMapEvent.Pair> datas = new ArrayList<TableMapEvent.Pair>();
        while (buffer.hasRemaining() && buffer.readed() < limit) {
            int col_index = (int) buffer.getPackedLength();
            int col_charset = (int) buffer.getPackedLength();

            Pair pair = new Pair();
            pair.col_index = col_index;
            pair.col_charset = col_charset;
            datas.add(pair);
        }

        return datas;
    }

    private List<Integer> parse_column_charset(Buffer buffer, int length) {
        // stores collation numbers extracted from field.
        int limit = buffer.readed() + length;
        List<Integer> datas = new ArrayList<Integer>();
        while (buffer.hasRemaining() && buffer.readed() < limit) {
            int col_charset = (int) buffer.getPackedLength();
            datas.add(col_charset);
        }

        return datas;
    }

    private void parse_column_name(Buffer buffer, int length) {
        // stores column names extracted from field
        int limit = buffer.readed() + length;
        int index = 0;
        while (buffer.hasRemaining() && buffer.readed() < limit) {
            int len = (int) buffer.getPackedLength();
            columnInfos[index++].name = buffer.getFixLengthStringWithNullTerminateCheck(len);
        }
    }

    private void parse_set_str_value(Buffer buffer, int length, boolean set) {
        // stores SET/ENUM column's string values extracted from
        // field. Each SET/ENUM column's string values are stored
        // into a string separate vector. All of them are stored
        // in 'vec'.
        int limit = buffer.readed() + length;
        List<List<String>> datas = new ArrayList<List<String>>();
        while (buffer.hasRemaining() && buffer.readed() < limit) {
            int count = (int) buffer.getPackedLength();
            List<String> data = new ArrayList<String>(count);
            for (int i = 0; i < count; i++) {
                int len1 = (int) buffer.getPackedLength();
                data.add(buffer.getFixLengthStringWithNullTerminateCheck(len1));
            }

            datas.add(data);
        }

        int index = 0;
        for (int i = 0; i < columnCount; i++) {
            if (set && getRealType(columnInfos[i].type, columnInfos[i].meta) == MysqlType.mysql_type_set) {
                columnInfos[i].set_enum_values = datas.get(index);
                index++;
            }

            if (!set && getRealType(columnInfos[i].type, columnInfos[i].meta) == MysqlType.mysql_type_enum) {
                columnInfos[i].set_enum_values = datas.get(index);
                index++;
            }
        }
    }

    private void parse_geometry_type(Buffer buffer, int length) {
        // stores geometry column's types extracted from field.
        int limit = buffer.readed() + length;

        List<Integer> datas = new ArrayList<Integer>();
        while (buffer.hasRemaining() && buffer.readed() < limit) {
            int col_type = (int) buffer.getPackedLength();
            datas.add(col_type);
        }

        int index = 0;
        for (int i = 0; i < columnCount; i++) {
            if (columnInfos[i].type == MysqlType.mysql_type_geometry) {
                columnInfos[i].geoType = datas.get(index);
                index++;
            }
        }
    }

    private void parse_simple_pk(Buffer buffer, int length) {
        // stores primary primaryKey's column information extracted from
        // field. Each column has an binlog_event_position_manager and a prefix which are
        // stored as a unit_pair. prefix is always 0 for
        // SIMPLE_PRIMARY_KEY field.

        int limit = buffer.readed() + length;
        while (buffer.hasRemaining() && buffer.readed() < limit) {
            int col_index = (int) buffer.getPackedLength();
            columnInfos[col_index].pk = true;
        }
    }

    private void parse_pk_with_prefix(Buffer buffer, int length) {
        // stores primary primaryKey's column information extracted from
        // field. Each column has an binlog_event_position_manager and a prefix which are
        // stored as a unit_pair.
        int limit = buffer.readed() + length;
        while (buffer.hasRemaining() && buffer.readed() < limit) {
            int col_index = (int) buffer.getPackedLength();
            // prefix length, 比如 char(32)
            @SuppressWarnings("unused")
            int col_prefix = (int) buffer.getPackedLength();
            columnInfos[col_index].pk = true;
        }
    }

    private boolean is_numeric_type(int type) {
        switch (type) {
            case MysqlType.mysql_type_tiny:
            case MysqlType.mysql_type_short:
            case MysqlType.mysql_type_int24:
            case MysqlType.mysql_type_long:
            case MysqlType.mysql_type_longlong:
            case MysqlType.mysql_type_newdecimal:
            case MysqlType.mysql_type_float:
            case MysqlType.mysql_type_double:
                return true;
            default:
                return false;
        }
    }

    private boolean is_character_type(int type) {
        switch (type) {
            case MysqlType.mysql_type_string:
            case MysqlType.mysql_type_var_string:
            case MysqlType.mysql_type_varchar:
            case MysqlType.mysql_type_blob:
                return true;
            default:
                return false;
        }
    }

    private int getRealType(int type, int meta) {
        if (type == MysqlType.mysql_type_string) {
            if (meta >= 256) {
                int byte0 = meta >> 8;
                if ((byte0 & 0x30) != 0x30) {
                    /* a long CHAR() field: see #37426 */
                    type = byte0 | 0x30;
                } else {
                    switch (byte0) {
                        case MysqlType.mysql_type_set:
                        case MysqlType.mysql_type_enum:
                        case MysqlType.mysql_type_string:
                            type = byte0;
                    }
                }
            }
        }

        return type;
    }


    public static final class ColumnInfo {

        public int type;
        public int meta;
        public String name;
        public boolean unsigned;
        public boolean pk;
        public List<String> set_enum_values;
        public int charset; // 可以通过CharsetUtil进行转化
        public int geoType;
        public boolean nullable;

        @Override
        public String toString() {
            return "ColumnInfo [eventType=" + type + ", meta=" + meta + ", taskId=" + name + ", unsigned=" + unsigned
                    + ", pk=" + pk + ", set_enum_values=" + set_enum_values + ", charset=" + charset + ", geoType="
                    + geoType + ", nullable=" + nullable + "]";
        }
    }

    private static final class Pair {

        public int col_index;
        public int col_charset;
    }

}
